package com.nokia.analytics.avrotest.AvroPositioningETL;

import java.io.IOException;

import org.apache.avro.mapreduce.AvroJob;
import org.apache.avro.mapreduce.AvroKeyOutputFormat;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nokia.analytics.AvroTypes.Position;

public class AvroPositioningBinaryDriver extends Configured implements Tool {

	public void printUsage() {
		System.out.println("Usage: hadoop jar path/to/this.jar "
				+ this.getClass() + " <input dir> <output dir>");

	}

	public int run(String[] args) throws IOException, InterruptedException,
			ClassNotFoundException {

		JobConf conf = new JobConf();
		String[] otherArgs = new GenericOptionsParser(conf, args)
				.getRemainingArgs();
		
		if (otherArgs.length != 2) {
			printUsage();
			System.exit(1);
		}

		Job job = new Job(conf, "Positioning binary convertion");
		job.setJarByClass(AvroPositioningBinaryDriver.class);

		FileInputFormat.setInputPaths(job, new Path(otherArgs[0]));
		Path outputPath = new Path(otherArgs[1]);
		FileOutputFormat.setOutputPath(job, outputPath);
		FileOutputFormat.setCompressOutput(job, true);
		FileOutputFormat.setOutputCompressorClass(job,
				org.apache.hadoop.io.compress.SnappyCodec.class);

		job.setInputFormatClass(TextInputFormat.class);

		job.setMapperClass(AvroPositioningBinaryMapper.class);

		AvroJob.setOutputKeySchema(job, new Position().getSchema());
		job.setOutputFormatClass(AvroKeyOutputFormat.class);

		job.setNumReduceTasks(0);

		return job.waitForCompletion(true) ? 0 : 1;
	}

	public static void main(String[] args) throws Exception {

		int res = ToolRunner.run(new Configuration(),
				new AvroPositioningBinaryDriver(), args);
		System.exit(res);

	}

}
